package com.imooc.log.`type`


import com.imooc.log.utils.{IPUtils, UAUtils}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.sources.{BaseRelation, TableScan}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}

import scala.collection.mutable.ArrayBuffer

/**
 * 自定义外部数据源解析逻辑
 */
class ImoocLogSourceRelation(override val sqlContext: SQLContext, path: String, userSchema: StructType)
  extends BaseRelation with TableScan with Logging with Serializable {


  override def schema: StructType = {
    if (userSchema != null) {
      userSchema
    } else {
      StructType(
        StructField("ip", StringType, false) ::
          StructField("city", StringType, false) ::
          StructField("time", StringType, false) ::
          StructField("method", StringType, false) ::
          StructField("url", StringType, false) ::
          StructField("protocal", StringType, false) ::
          StructField("status", StringType, false) ::
          StructField("bytessent", StringType, false) ::
          StructField("referer", StringType, false) ::
          StructField("ua", StringType, false) ::
          StructField("browsername", StringType, false) ::
          StructField("browserversion", StringType, false) ::
          StructField("osname", StringType, false) ::
          StructField("osversion", StringType, false) :: Nil
      )
    }
  }

  /**
   * 把数据读进来，读进来之后把它转换成 RDD[Row]
   *
   * @return
   */
  override def buildScan(): RDD[Row] = {
    logWarning("this is ruozedata buildScan....")
    //读取数据，变成为RDD
    //wholeTextFiles会把文件名读进来，可以通过map(_._2)把文件名去掉，第一位是文件名，第二位是内容
    val fileRdd = sqlContext.sparkContext.wholeTextFiles(path).map(_._2)
    val schemaFields = schema.fields

    val rows = fileRdd.map(fileContent => {
      // 取每一行
      val filelines = fileContent.split("\n")

      /**
       * 110.85.18.234 - - [30/Jan/2019:00:00:21 +0800]
       * GET /course/list?c=cb HTTP/1.1
       * 200 12800
       * www.imooc.com
       *
       * https://www.imooc.com/course/list?c=data
       * -
       * Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 SE 2.X MetaSr 1.0
       *
       * -
       * 10.100.16.243:80 200 0.172 0.172
       */
      val lines = logParse(filelines)

      val data = lines.map(line => line.split("\\^").toSeq)

      val tmp = data.map(words =>
        words.zipWithIndex.map({
          case (value, index) =>

            castTo(value, schemaFields(index).dataType)
        })
      )

      tmp.map(s => Row.fromSeq(s))
    })

    rows.flatMap(e => e)
  }

  def castTo(value: String, dataType: DataType) = {
    dataType match {
      case _: IntegerType => value.toInt
      case _: LongType => value.toLong
      case _: StringType => value.toString
    }
  }

  def logParse(fileline: Array[String]): Array[String] = {
    val res = new ArrayBuffer[String]()
    val str = fileline.map(data => {
      val tmp = data.split("\"").map(str => str.trim)
      val tmpA = tmp(0).split(" ")
      val ip = tmpA(0)
      val city = IPUtils.getCity(ip)
      val time = tmpA(3) + " " + tmpA(4)

      val tmpB = tmp(1).split(" ")
      val method = tmpB(0)
      val url = tmpB(1)
      val protocal = tmpB(2) // 协议

      val tmpC = tmp(2).split(" ")

      val status = tmpC(0) // 状态码
      val bytessent = tmpC(1) // 流量


      val referer = tmp(5)
      val ua = tmp(7)
      val info = UAUtils.getUserAgentInfo(ua)
      val browsername = info.getBrowserName
      val browserversion = info.getBrowserVersion
      val osname = info.getOsName
      val osversion = info.getOsVersion

      (new StringBuilder)
        .append(ip).append("^")
        .append(city).append("^")
        .append(time).append("^")
        .append(method).append("^")
        .append(url).append("^")
        .append(protocal).append("^")
        .append(status).append("^")
        .append(bytessent).append("^")
        .append(referer).append("^")
        .append(ua).append("^")
        .append(browsername).append("^")
        .append(browserversion).append("^")
        .append(osname).append("^")
        .append(osversion).toString
    })
    res ++= str
    res.toArray
  }
}